// 通过信号量实现“生产者-消费者模型”
#include <stdio.h>
#include <unistd.h>
#include <vector>
#include <pthread.h>
#include <semaphore.h>

#define CAPACITY 1  //资源队列大小 
#define THREADCOUNT 2  //每种线程数量

using namespace std;

/* 1.定义一个线程安全的队列
 *     队列：环形队列（数组模拟）
 *     线程安全：
 *         同步：信号量实现
 *         互斥：信号量实现
 * */
class SafeQueue
{
    public:
        // 构造
        SafeQueue()
            : _capacity(CAPACITY)
            , _vec(CAPACITY)
        {
            // 初始化信号量
            sem_init(&sem_mutex, 0, 1);  //互斥信号量（初始资源数为1）
            sem_init(&sem_cons_cond, 0, 0);  //消费者信号量（初始资源数为0）
            sem_init(&sem_prod_cond, 0, CAPACITY);  //生产者信号量（初始资源为队列长度）

            _pos_write = 0;
            _pos_read = 0;
        }
        // 析构
        ~SafeQueue()
        {
            // 销毁信号量
            sem_destroy(&sem_mutex);
            sem_destroy(&sem_cons_cond);
            sem_destroy(&sem_prod_cond);
        }
        // 消费者线程的POP接口
        void Pop()
        {
            sem_wait(&sem_cons_cond);
            sem_wait(&sem_mutex);

            int cur = _vec[_pos_read];
            _pos_read = (_pos_read + 1)%_capacity;  //更新_pos_read
            printf("Consume:%p get:%d\n", pthread_self(), cur);

            sem_post(&sem_prod_cond);
            sem_post(&sem_mutex);
        }
        // 生产者线程的PUSH接口
        void Push(int data)
        {
            sem_wait(&sem_prod_cond);
            sem_wait(&sem_mutex);

            printf("Product:%p make:%d\n", pthread_self(), data);
            _vec[_pos_write] = data;
            _pos_write = (_pos_write + 1)%_capacity;

            sem_post(&sem_cons_cond);
            sem_post(&sem_mutex);
            
        }

    private:
        size_t _capacity;  //数组容量
        vector<int> _vec;  //环形队列数组
        
        sem_t sem_mutex;  //队列资源互斥信号量
        sem_t sem_cons_cond;  //消费者线程的同步信号量
        sem_t sem_prod_cond;  //生产者线程的同步信号量

        int _pos_write;  //队列末尾，插入元素位置
        int _pos_read;  //队列头部，读取元素位置
};


// 消费者线程入口函数
void* cons_strat(void* arg)
{
    SafeQueue* sq = (SafeQueue*)arg;
    while(1)
    {
        sq->Pop();
        sleep(2);
    }
}

//生产者向循环队列存放的数据
int g_data = 0;  
//设置静态互斥锁，不同生产者对g_data必须互斥访问
pthread_mutex_t g_lock = PTHREAD_MUTEX_INITIALIZER;

// 生产者线程入口函数
void* prod_strat(void* arg)
{
    SafeQueue* sq = (SafeQueue*)arg;
    while(1)
    {
        pthread_mutex_lock(&g_lock);
        sq->Push(g_data++);
        pthread_mutex_unlock(&g_lock);
        sleep(1);
    }
}

// 创建生产者-消费者两种线程
int main(){
    SafeQueue* sq = new SafeQueue();

    pthread_t cons[THREADCOUNT], prod[THREADCOUNT];
    for(int i = 0; i < THREADCOUNT; i++){
        int ret = pthread_create(&cons[i], NULL, cons_strat, (void*)sq);
        if(ret < 0)
        {
            perror("pthread_create");
            return 0;
        }
        
        ret = pthread_create(&prod[i], NULL, prod_strat, (void*)sq);
        if(ret < 0)
        {
            perror("pthread_create");
            return 0;
        }
    }

    for(int i = 0; i < THREADCOUNT; i++)
    {
        pthread_join(cons[i], NULL);
        pthread_join(prod[i], NULL);
    }
    delete sq;
    return 0;
}
